package com.pxene.caw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class ClearDataTask extends Configuration  implements Tool   {

	private String outputPath="/user/vendorwins/chenjinghui";
	private Path totalPath=new Path("/user/vendorwins/chenjinghui");
	
	@Override
	public void setConf(Configuration conf) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Configuration getConf() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf=new Configuration();
		conf.set("mapreduce.job.queuename", "ven1"); //设置队列
		FileSystem fs=FileSystem.get(conf);
		if(fs.exists(totalPath)){
			fs.delete(totalPath, true);
		}
		Job job=Job.getInstance(conf);
		job.setJarByClass(ClearDataTask.class);
		job.setNumReduceTasks(0);
		job.setMapperClass(ClearDataMapper.class);
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		
		Path bj=new Path("/daas/bstl/dpiqixin/beijing/20151218");
		Path bjPath=new Path("/daas/bstl/dpiqixin/beijing/20151115/MBLDPI.2015111523.7.1447602727877.lzo_deflate");
		Path hbPath1=new Path("/daas/bstl/dpiqixin/hubei/20151104/MBLDPI.2015110423.6.1446650251949.lzo_deflate");
		Path hbPath2=new Path("/daas/bstl/dpiqixin/hubei/20151104/MBLDPI.2015110423.6.1446652069152.lzo_deflate");
		Path hbpath3=new Path("/daas/bstl/dpiqixin/hubei/20151104/MBLDPI.2015110423.7.1446649201065.lzo_deflate");
		Path shanghai1=new Path("/daas/bstl/dpiqixin/shandong/20151115/MBLDPI.2015111523.1447599600115.lzo_deflate");
		Path shanghai2=new Path("/daas/bstl/dpiqixin/shandong/20151115/MBLDPI.2015111523.1447599600104.lzo_deflate");
		Path shanghai3=new Path("/daas/bstl/dpiqixin/shandong/20151115/MBLDPI.2015111523.1447599600090.lzo_deflate");
		Path shanghai4=new Path("/daas/bstl/dpiqixin/sichuan/20151123/MBLDPI.2015112323.1448290800129.lzo_deflate");
		Path shanghai5=new Path("/daas/bstl/dpiqixin/sichuan/20151123/MBLDPI.2015112323.1448290800134.lzo_deflate");
		Path shanghai6=new Path("/daas/bstl/dpiqixin/sichuan/20151123/MBLDPI.2015112323.1448290800108.lzo_deflate");
		
		
		SequenceFileInputFormat.setInputPaths(job,new Path(args[0]));
		
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
//		if(fs.exists(new Path(outputPath))){
//			fs.delete(new Path(outputPath), true);
//		}
		return  job.waitForCompletion(true)?1:0;
	}
		
		
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new ClearDataTask(), args);
	}
}
